package com.hibase.core.mq.redis.consumer;

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.hibase.common.constant.RedisKeyConstant;
import com.hibase.common.util.DecimalUtil;
import com.hibase.core.mq.annotation.HibaseDelayQueueConsumer;
import com.hibase.core.mq.consumer.IQueueConsumer;
import com.hibase.core.redis.util.RedisUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.ZSetOperations;

import javax.annotation.PostConstruct;
import java.math.BigDecimal;
import java.util.Set;
import java.util.concurrent.Executors;

/**
 * @author 杨斌冰-工具组-技术中心
 *         <p>
 *         2018/3/1 15:31
 */
@Slf4j
public abstract class AbstractRedisDelayQueueConsumer<T> implements IQueueConsumer {

    @Autowired
    private RedisUtil redisUtil;

    private String queueName;

    public abstract void doConsume(T message);

    @Override
    public void consume() throws Exception{
        if(this.getClass().isAnnotationPresent(HibaseDelayQueueConsumer.class)){
            HibaseDelayQueueConsumer delayQueueConsumer = this.getClass().getAnnotation(HibaseDelayQueueConsumer.class);
            queueName = StrUtil.isNotBlank(delayQueueConsumer.value()) ? delayQueueConsumer.value() : delayQueueConsumer.queueName();
            if(StrUtil.isBlank(queueName)){
                throw new Exception("Redis Delay Queue Consumer Java Config Error.");
            }
            while (true){
                Set<ZSetOperations.TypedTuple<Object>> result = redisUtil.zRangeWithScores(RedisKeyConstant.DELAY_QUEUE_PREFIX_KEY + queueName, 0, 0);
                Thread.sleep(100);
                if(ObjectUtil.isNotEmpty(result) && !result.isEmpty()){
                    //判断元素是否超时  根据超时时间戳 单位全部是毫秒
                    ZSetOperations.TypedTuple<Object> tuple = (ZSetOperations.TypedTuple<Object>)result.toArray()[0];
                    // 超时时间戳
                    BigDecimal score = new BigDecimal(tuple.getScore());
                    BigDecimal current = new BigDecimal(System.currentTimeMillis());
                    // 超时
                    if(DecimalUtil.ge(current,score)){
                        // 删除
                        Long zRemove = redisUtil.zRemove(RedisKeyConstant.DELAY_QUEUE_PREFIX_KEY + queueName, tuple.getValue());
                        if(ObjectUtil.isNotEmpty(zRemove) && zRemove > 0){
                            this.doConsume(JSON.parseObject((String) tuple.getValue(),new TypeReference<T>(){}));
                        }
                    }
                    else {
                        Thread.sleep(100);
                    }
                }
            }
        }else{
            throw new Exception("Redis Delay Queue Consumer Java Config Error.");
        }
    }

    @PostConstruct
    public void init(){
        log.info("The Consumer in Delay Queue [{}] Started.",this.getClass().getName());
        Executors.newSingleThreadExecutor().submit(()->{
            try {
                this.consume();
            } catch (Exception e) {
                log.error("The Consumer in Delay Queue [" + this.queueName + "] Error.",e);
            }
        });
    }
}
